package com.andox.fn.inf.rpc.rest.common;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.andox.fn.inf.rpc.rest.RpcRequest;
import com.andox.fn.inf.rpc.rest.RpcResponse;
import com.andox.fn.inf.rpc.rest.RemotingCallback;
import com.andox.fn.inf.rpc.rest.RequestContext;
import com.andox.fn.inf.rpc.rest.RequestContextConstants;
import com.andox.fn.inf.rpc.rest.ResponseCodes;
import com.andox.fn.inf.rpc.rest.exception.TimeoutException;

/**   
 * @ClassName:   ResponseFuture   
 * @author:      Andox
 * @date:        2019年8月1日 下午2:33:10   
 *     
 */
public class ResponseFuture {
	private final static Logger logger = LoggerFactory.getLogger(ResponseFuture.class);
	
	private final static Map<Long, ResponseFuture> futureTable = new ConcurrentHashMap<>();
	
	private CountDownLatch countDownLatch = new CountDownLatch(1);
	private RpcRequest request;
	private RpcResponse response;
	
	private Exception exception;
	
	private long startTimestampInMilliseconds;
	private int timeoutInMilliseconds;
	
	private RemotingCallback callback;
	
	private RequestContext requestContext = RequestContext.instance();
	
	public ResponseFuture(RpcRequest request) {
		this.request = request;
		futureTable.put(request.getOpaque(), this);
	}
	
	public ResponseFuture() {

	}
	
	public RpcResponse get(int timeInMilliseconds) {
		try {
			boolean success = this.countDownLatch.await(timeInMilliseconds, TimeUnit.MILLISECONDS);
			if (success) {
				return response;
			}
		} catch (Exception e) {
			
		}
		
		throw new TimeoutException("timeout");
	}
	
	public boolean isDone() {
		return this.response != null;
	}
	
	public static void putResponse(RpcResponse response) {
		ResponseFuture rf = futureTable.get(response.getOpaque());
		
		if (rf == null) {
			logger.error("recieved error response, or timeout., response id: {}", response.getOpaque());
			return;
		}
		
		
		rf.requestContext.putProperty(RequestContextConstants.BIZ_RESPONSE, response);
		rf.response = response;
		futureTable.remove(response.getOpaque());
		rf.countDownLatch.countDown();
		
		if (rf.callback != null) {
			if (rf.response.getCode().equals(ResponseCodes.SUCCESS)) {
				rf.callback.onSuccess(response);
			}else {
				rf.callback.onFailed(null);
			}
		}
	}
	
	/**
	 * @return the request
	 */
	public RpcRequest getRequest() {
		return request;
	}

	/**
	 * @param request the request to set
	 */
	public void setRequest(RpcRequest request) {
		this.request = request;
	}

	/**
	 * @return the startTimestampInMilliseconds
	 */
	public long getStartTimestampInMilliseconds() {
		return startTimestampInMilliseconds;
	}

	/**
	 * @param startTimestampInMilliseconds the startTimestampInMilliseconds to set
	 */
	public void setStartTimestampInMilliseconds(long startTimestampInMilliseconds) {
		this.startTimestampInMilliseconds = startTimestampInMilliseconds;
	}

	/**
	 * @return the timeoutInMilliseconds
	 */
	public int getTimeoutInMilliseconds() {
		return timeoutInMilliseconds;
	}

	/**
	 * @param timeoutInMilliseconds the timeoutInMilliseconds to set
	 */
	public void setTimeoutInMilliseconds(int timeoutInMilliseconds) {
		this.timeoutInMilliseconds = timeoutInMilliseconds;
	}

	/**
	 * @return the exception
	 */
	public Exception getException() {
		return exception;
	}

	/**
	 * @param exception the exception to set
	 */
	public void setException(Exception exception) {
		this.exception = exception;
	}

	private static class RemotingInvocationTimeoutScan implements Runnable {

        public void run() {
            while (true) {
                try {
                    for (ResponseFuture future : futureTable.values()) {
                        if (future == null || future.isDone()) {
                            continue;
                        }
                        
                        int del = 60;
                        if (System.currentTimeMillis() - future.startTimestampInMilliseconds > (del + future.timeoutInMilliseconds)) {
                        	logger.error("remove timeout response future. request id: {}, current: {}, start: {}, timeout: {}", 
                        			future.request.getOpaque(), System.currentTimeMillis(), future.startTimestampInMilliseconds, future.timeoutInMilliseconds);
                            RpcResponse timeoutResponse = new RpcResponse(future.request.getOpaque());
                            timeoutResponse.setOpaque(future.request.getOpaque());
                            timeoutResponse.setCode(ResponseCodes.TIMEOUT_ERROR);
                            timeoutResponse.setMsg("time out");
                            logger.error("remove timeout response future. request id: {}", future.request.getOpaque());
                            ResponseFuture.putResponse(timeoutResponse);
                        }
                    }
                    Thread.sleep(30);
                } catch (Throwable e) {
                    logger.error("Exception when scan the timeout invocation of remoting.", e);
                }
            }
        }
    }

    static {
        Thread th = new Thread(new RemotingInvocationTimeoutScan(), "ResponseTimeoutScanTimer");
        th.setDaemon(true);
        th.start();
    }
}
